package stream;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

public class operator_split {
    public static void main(String[] args) throws Exception {
    //   创建流式数据环
        StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();
    //    添加数据源
        DataStreamSource<Long> source = env.addSource(new oneparalleSource());

        SplitStream<Long> data =  source.split(new OutputSelector<Long>() {
            public Iterable<String> select(Long aLong) {
                //定义一个list 分成几个流
                //定义一个List来保存切分后，流的个数（奇数流、偶数流）
                //在该方法中，给数据加上一个或者多个标识符
                ArrayList<String> selector = new ArrayList<String>();
                if (aLong % 2 == 0){
                    selector.add("o");
                    selector.add("i");
                }
                else{
                    selector.add("j");
                }
                return selector;
            }
        });
        //result.timeWindowAll(Time.seconds(2)).sum(0);
        //result.print().setParallelism(2);
        DataStream<Long> result = data.select("j");
        DataStream<Long> result2 = data.select("i");
        result.print();
        result2.print();
        env.execute("operatro_filter");
    }
}

